package com.flink.windowFunctionDemo;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class FullWindowFunctionExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, Integer>> inputStream = env.fromElements(
                new Tuple2<>("key1", 1),
                new Tuple2<>("key1", 2),
                new Tuple2<>("key2", 3)
        );

        DataStream<Tuple2<String, Integer>> resultStream = inputStream
                .keyBy(value -> value.f0)
                .timeWindow(Time.seconds(5))
                .process(new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
                        int sum = 0;
                        for (Tuple2<String, Integer> element : elements) {
                            sum += element.f1;
                        }
                        out.collect(new Tuple2<>(key, sum));
                    }
                });

        resultStream.print();
        env.execute("Full Window Function Example");
    }
}

/*
全量窗口函数（Full Window Functions）
介绍：需要将窗口内的所有数据都收集起来，等窗口触发时再进行计算。这种方式会占用较多的内存，但可以实现更复杂的计算。
示例：WindowFunction 和 ProcessWindowFunction 属于全量窗口函数。*/
